
import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import scala.Tuple2;

import java.io.IOException;
import java.util.*;

/**
 * @author <a href="mailto:xingxiao@gtmap.cn">xingxiaofeng</a>
 * @version2.1 2022-03-10
 * @description
 */
public class WordCountKafKaHbaseStram {
    static Configuration conf = null;
    static {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "h1,h2,h3");
    }
    public static void main(String[] args) throws InterruptedException {
        String brokers = "h2:9092,h3:9092,h4:9092";
        String topics = "spark-streaming-topic";
        SparkConf conf = new SparkConf().setAppName("streaming word count");//.setMaster("local[2]");
        JavaSparkContext sc = new JavaSparkContext(conf);
        sc.setLogLevel("WARN");
        JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(1));

        Collection<String> topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));
        //kafka相关参数，必要！缺了会报错
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", brokers) ;
        kafkaParams.put("bootstrap.servers", brokers);
        kafkaParams.put("group.id", "group1");
        kafkaParams.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //Topic分区  也可以通过配置项实现
        //如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性
        //earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
        //latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
        //none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        //kafkaParams.put("auto.offset.reset", "latest");
        //kafkaParams.put("enable.auto.commit",false);

        Map offsets=new HashMap<>();
        offsets.put(new TopicPartition("spark-streaming-topic", 0), 2L);
        //通过KafkaUtils.createDirectStream(...)获得kafka数据，kafka相关参数由kafkaParams指定
        JavaInputDStream<ConsumerRecord<Object,Object>> lines = KafkaUtils.createDirectStream(
                ssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topicsSet, kafkaParams, offsets)
        );
        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.value().toString().split(" ")).iterator());


        JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        });
        pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer+integer2;
            }
        }).foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {


            @Override
            public void call(JavaPairRDD<String, Integer> pairRDD) throws Exception {

                pairRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {

                    @Override
                    public void call(Tuple2<String, Integer> tuple2) throws Exception {

                        add("tb1","100","cf",tuple2._1,tuple2._2.toString());
                    }
                });

            }
        });

        //这里就跟之前的demo一样了，只是需要注意这边的lines里的参数本身是个ConsumerRecord对象
/*        JavaPairDStream<String, Integer> counts =
                lines.flatMap(x -> Arrays.asList(x.value().toString().split(" ")).iterator())
                        .mapToPair(x -> new Tuple2<String, Integer>(x, 1))
                        .reduceByKey((x, y) -> x + y);
        counts.print();*/
//  可以打印所有信息，看下ConsumerRecord的结构
//      lines.foreachRDD(rdd -> {
//          rdd.foreach(x -> {
//            System.out.println(x);
//          });
//        });
        ssc.start();
        ssc.awaitTermination();
        ssc.close();


    }

    public static void add(String tableName,String rowKey,String family,String qualifier,String value) throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));
        HTable table = new HTable(conf, tableName);
        put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
        table.put(put);
        System.out.println("key:"+qualifier+";count:"+value);
        System.out.println("add data success");
        System.out.println("===========");
    }
    public static void ad(String key,String value){
        String tablebName = "tb12";
        String rowKey = "xiaoming";
        String[] family = {"cf","address","score"};
        try {
//          createTable(tablebName, family);
            add(tablebName,rowKey,family[0],"age","18");
            add(tablebName,rowKey,family[0],"birthday","1990-12-12");
            add(tablebName,rowKey,family[0],"school","beijingdaxue");
            add(tablebName,rowKey,family[1],"country","china");
            add(tablebName,rowKey,family[1],"province","guangdong");
            add(tablebName,rowKey,family[1],"city","shenzhen");
            add(tablebName,rowKey,family[2],"yuwen","99");
            add(tablebName,rowKey,family[2],"shuxue","98");
            add(tablebName,rowKey,family[2],"yingyu","100");
        } catch (MasterNotRunningException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}
